﻿using Confluent.Kafka;
using Confluent.Kafka.SyncOverAsync;
using Confluent.SchemaRegistry;
using Confluent.SchemaRegistry.Serdes;
using kafkalearn.ProducerTest1.AvroSpecific;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ConsumerTest1
{
    public class AvroManualAssignConsumer : IConsumer
    {
        private readonly int _partition;
        private readonly string _schemaUri;
        private readonly string _topic;
        public AvroManualAssignConsumer(string schemaUri, string topic, int partition)
        {
            _topic = topic;
            _schemaUri = schemaUri;
            _partition = partition;
        }
        public Task ExcuteAsync(CancellationToken token = default)
        {
            SchemaRegistryConfig schemaRegistryConfig = new SchemaRegistryConfig
            {
                Url = _schemaUri
            };
            ConsumerConfig config = new ConsumerConfig();
            config.BootstrapServers = "192.168.3.68:9092,192.168.3.66:9092,192.168.3.69:9092";
            config.GroupId = "charp-consumer";
            config.EnableAutoCommit = true;
            using(var schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig))
            {
                using (IConsumer<string, User> consumer = new ConsumerBuilder<string, User>(config)
                .SetErrorHandler((_, e) => Console.WriteLine($"Error:{e.Reason}"))
                .SetKeyDeserializer(new AvroDeserializer<string>(schemaRegistry).AsSyncOverAsync())
                .SetValueDeserializer(new AvroDeserializer<User>(schemaRegistry).AsSyncOverAsync())
                .SetPartitionsAssignedHandler((_, partitions) =>
                {
                    Console.WriteLine($"Incremental partition assignment:[{string.Join(", ", partitions)}]");
                })
                .Build())
                {
                    //从指定主题指定分区的最开始位置读起
                    consumer.Assign(new TopicPartitionOffset(_topic, _partition, Offset.Beginning));
                    bool r = false;
                    try
                    {
                        while (true)
                        {
                            try
                            {
                                var consumerResult = consumer.Consume(token);
                                Console.WriteLine($"Received message at {consumerResult.TopicPartitionOffset}: {consumerResult.Message.Value.name}");
                                TopicPartition tp = new TopicPartition(consumerResult.Topic, consumerResult.Partition);
                                List<TopicPartition> _pause = new List<TopicPartition>();
                                _pause.Add(tp);
                                if (!r)
                                {
                                    consumer.Pause(_pause); //这个仅会暂停接收消息，但是不会阻塞轮询
                                }
                                Task.Run(() => {
                                    Thread.Sleep(10000);
                                    consumer.Resume(_pause);
                                    r = true;
                                });
                            }
                            catch (ConsumeException e)
                            {
                                Console.WriteLine($"Consume error: {e.Error.Reason}");
                            }
                        }
                    }
                    catch (OperationCanceledException e)
                    {
                        Console.WriteLine("Closing consumer.");
                        consumer.Close();
                    }
                }
            }
            
            return Task.CompletedTask;
        }
    }
}
